package xnet

import (
	"fmt"
	"gitee.com/shuohe/xutils/xlog"
	"io"
	"net"
	"sync"
	"time"

	. "gitee.com/shuohe/xutils"
	"gitee.com/shuohe/xutils/xerror"
)

const (
	MAX_MESSAGE_SIZE = 8 * 1024
)

//=============================================================================

func errAlreadyClosed() error {
	return xerror.SimpleNew(Caller(2) + ": Channel is already closed.")
}

func errInvalidMessageLength() error {
	return xerror.SimpleNew(Caller(2) + ": Got an invalid message length.")
}

func errRead(err error, what string) error {
	return xerror.SimpleWrapf(err, Caller(2)+": Read %s failed.", what)
}

func errWrite(err error, what string) error {
	return xerror.SimpleWrapf(err, Caller(2)+": Write %s failed.", what)
}

type TCPChannel struct {
	conn *net.TCPConn

	// 事件处理器
	callback ITCPChannelCallback

	// 上次收到消息和发送消息时间
	lastRead  time.Time
	lastWrite time.Time
	// 读写两个消息之间的最大时间间隔
	// 可读写
	IdleTimeout time.Duration

	// 用于接收读取到数据的缓存，避免每次动态分配内存
	rbuf []byte
	// 在读消息时，是否重复使用预分配的 rbuf
	// 可读写
	ReuseBuffer bool

	rmux sync.Mutex
	wmux sync.Mutex

	// 所属的 TCPServer
	// 可读写，一般在刚创建好时设置
	Server *TCPServer
	// 连接在 TCPServer 内的唯一标识
	// 可读写，一般在刚创建好时设置
	Id int

	// 指示是否已经进入伺服状态
	// 只读
	IsServing bool
	// 指示连接是否已经被关闭了
	// 只读
	IsClosed bool
}

func (this *TCPChannel) preventWriteTimeout() {
	for this.IsServing {
		time.Sleep(time.Second)

		if this.IsClosed || !this.IsServing {
			break
		}

		// 每隔一秒检查一次是否快要超时
		if time.Now().Sub(this.lastWrite) < (this.IdleTimeout - time.Second) {
			continue
		}

		err := WriteFull(this.conn, []byte{0, 0, 0, 0})
		if err != nil {
			xlog.Errorfe(err, "Failed to send keep-alive message.")
			this.callback.OnError(err)
			// 跳出循环，同时也通知 Serve() 例程退出，从而关闭连接
			this.IsServing = false
		}
		this.lastWrite = time.Now()
	}
}

func (this *TCPChannel) FullName() string {
	return fmt.Sprintf("Channel-(%v)", this.conn.RemoteAddr())
}

func (this *TCPChannel) RemoteAddr() *net.TCPAddr {
	return this.conn.RemoteAddr().(*net.TCPAddr)
}

func (this *TCPChannel) SetCallback(cb ITCPChannelCallback) {
	this.callback = cb
	if this.callback != nil {
		this.callback.SetChannel(this)
	}
}

func (this *TCPChannel) SetDeadline(deadline time.Time) {
	this.conn.SetDeadline(deadline)
}

func (this *TCPChannel) SetBufferSize(size int) {
	this.rbuf = make([]byte, size)
}

func (this *TCPChannel) ClearDeadline() {
	this.conn.SetDeadline(time.Time{})
}

func (this *TCPChannel) Serve() {
	if this.IsClosed || this.IsServing || this.callback == nil {
		return
	}

	this.IsServing = true

	this.lastRead = time.Now()
	this.lastWrite = time.Now()
	if this.IdleTimeout > 0 {
		go this.preventWriteTimeout()
	}

	for this.IsServing {
		if this.IdleTimeout > 0 {
			this.conn.SetReadDeadline(this.lastRead.Add(this.IdleTimeout))
		}
		// 阻塞，直到出错或读到完整的消息
		msg, err := this.Read()
		if err != nil {
			this.callback.OnError(err)
			this.IsServing = false
			break
		}

		if msg == nil {
			// this.Read() 返回 (nil, nil) 意味着刚读到一个 keep-alive 消息
			continue
		}
		// 读到正常消息，回调
		err = this.callback.OnMessage(msg)
		if err != nil {
			xlog.Errorfe(err, "OnMessage callback failed.")
			break
		}
	}

	// 读操作出现错误，或者 Close() 方法被调用了
	this.close()
}

func (this *TCPChannel) Read() ([]byte, error) {
	if this.IsClosed {
		return nil, errAlreadyClosed()
	}

	this.rmux.Lock()
	defer this.rmux.Unlock()

	l, err := ReadUint32(this.conn)
	if err != nil {
		// 读操作失败
		return nil, err
	} else if l == 0 {
		// 读到一个长度为零的 keep-alive 消息
		return nil, nil
	}

	var msg []byte

	if this.ReuseBuffer {
		if l > uint32(len(this.rbuf)) {
			return nil, errInvalidMessageLength()
		}
		msg = this.rbuf[:l]
	} else {
		if l > MAX_MESSAGE_SIZE {
			return nil, errInvalidMessageLength()
		}
		msg = make([]byte, l)
	}

	// 读取完整消息
	_, err = io.ReadFull(this.conn, msg)
	if err != nil {
		return nil, errRead(err, "message body")
	}

	this.lastRead = time.Now()

	return msg, nil
}

func (this *TCPChannel) Write(msg []byte) error {
	if this.IsClosed {
		return errAlreadyClosed()
	}

	this.wmux.Lock()
	defer this.wmux.Unlock()

	err := WriteUint32(this.conn, uint32(len(msg)))
	if err != nil {
		return errWrite(err, "message head")
	}

	err = WriteFull(this.conn, msg)
	if err != nil {
		return errWrite(err, "message body")
	}

	this.lastWrite = time.Now()

	return nil
}

func (this *TCPChannel) close() {
	if this.callback != nil {
		this.callback.OnClose()
	}

	if this.Server != nil {
		delete(this.Server.Channels, this.Id)
	}

	this.conn.Close()
	this.IsClosed = true
}

// 通道对象关闭后，就不能再次使用了，必须重新创建通道。
func (this *TCPChannel) Close() {
	if this.IsClosed {
		return
	}

	if this.IsServing {
		// 设置关闭标志，让 Serve() 函数跳出循环，从而真正关闭连接
		this.IsServing = false
	} else {
		this.close()
	}
	this.conn.Close()
}

func Connect(remoteAddr *net.TCPAddr, callback ITCPChannelCallback) (*TCPChannel, error) {
	conn, err := net.DialTCP("tcp4", nil, remoteAddr)
	if err != nil {
		return nil, xerror.Wrap(err, "Can not connect to the peer.")
	}

	chnl := NewTCPChannel(conn, callback)
	if callback != nil {
		err := chnl.callback.OnConnect()
		if err != nil {
			chnl.Close()
			return nil, xerror.Wrap(err, "OnConnect() failed.")
		}
	}

	return chnl, nil
}

/**
 * 创建 TCPChannel 对象。
 *
 * 对象的 Server、Id 字段需要另行赋值。
 */
func NewTCPChannel(conn *net.TCPConn, callback ITCPChannelCallback) *TCPChannel {
	chnl := &TCPChannel{
		conn: conn,
	}
	chnl.SetCallback(callback)

	// 默认在读消息时，使用预分配的缓冲区
	chnl.SetBufferSize(MAX_MESSAGE_SIZE)
	chnl.ReuseBuffer = true

	return chnl
}
